package com.ksyun.campus.metaserver.storage;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.ksyun.campus.metaserver.config.MinfsProperties;
import com.ksyun.campus.metaserver.domain.*;
import lombok.extern.slf4j.Slf4j;
import org.rocksdb.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.File;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;

/**
 * RocksDB implementation of metadata store
 * 元数据存储的RocksDB实现
 * 使用RocksDB作为底层存储引擎，提供高性能的键值存储服务
 * 实现了MetadataStore接口定义的所有元数据操作
 */
@Slf4j
@Component
public class RocksDBMetadataStore implements MetadataStore {
    private static final String INODE_PREFIX = "inode:";
    private static final String DENTRY_PREFIX = "dentry:";
    private static final String PATH_CACHE_PREFIX = "pathcache:";
    private static final String BLOCK_PREFIX = "block:";
    private static final String DS_PREFIX = "ds:";
    private static final String CONF_PREFIX = "conf:";
    private static final String CLUSTER_KEY = "cluster";

    @Autowired
    private MinfsProperties properties;
    @Autowired
    private ObjectMapper objectMapper;
    private RocksDB db;
    private Options options;

    @Override
    public void start() throws Exception {
        log.info("Starting RocksDB metadata store");
        try {
            RocksDB.loadLibrary();
            options = new Options();
            options.setCreateIfMissing(true);
            options.setWriteBufferSize(properties.getMetadata().getRocksdb().getWriteBufferSize());
            options.setMaxWriteBufferNumber(properties.getMetadata().getRocksdb().getMaxWriteBufferNumber());
            options.setLevel0FileNumCompactionTrigger(properties.getMetadata().getRocksdb().getLevel0FileNumCompactionTrigger());
            options.setMaxBytesForLevelBase(properties.getMetadata().getRocksdb().getMaxBytesForLevelBase());
            String dbPath = properties.getMetadata().getRocksdb().getPath();
            File dbDir = new File(dbPath);
            if (!dbDir.exists()) {
                boolean created = dbDir.mkdirs();
                if (!created) {
                    throw new RuntimeException("Failed to create RocksDB directory: " + dbPath);
                }
            }
            db = RocksDB.open(options, dbPath);
            if (db == null) {
                throw new RuntimeException("Failed to open RocksDB at: " + dbPath);
            }
            log.info("RocksDB metadata store started at: {}", dbPath);
        } catch (Exception e) {
            log.error("Failed to start RocksDB metadata store", e);
            if (db != null) {
                try {
                    db.close();
                } catch (Exception closeEx) {
                    log.error("Error closing RocksDB", closeEx);
                }
                db = null;
            }
            if (options != null) {
                try {
                    options.close();
                } catch (Exception closeEx) {
                    log.error("Error closing RocksDB options", closeEx);
                }
                options = null;
            }
            throw e;
        }
    }

    @Override
    public void stop() throws Exception {
        log.info("Stopping RocksDB metadata store");
        if (db != null) {
            db.close();
        }
        if (options != null) {
            options.close();
        }
        log.info("RocksDB metadata store stopped");
    }

    private void checkDbInitialized() throws Exception {
        if (db == null) {
            throw new IllegalStateException("RocksDB not initialized. Please call start() first.");
        }
    }

    @Override
    public void putInode(Inode inode) throws Exception {
        checkDbInitialized();
        String key = INODE_PREFIX + inode.getFileId();
        String value = objectMapper.writeValueAsString(inode);
        db.put(key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8));
    }

    @Override
    public Optional<Inode> getInode(String fileId) throws Exception {
        String key = INODE_PREFIX + fileId;
        byte[] value = db.get(key.getBytes(StandardCharsets.UTF_8));
        if (value == null) {
            return Optional.empty();
        }
        Inode inode = objectMapper.readValue(new String(value, StandardCharsets.UTF_8), Inode.class);
        return Optional.of(inode);
    }

    @Override
    public void deleteInode(String fileId) throws Exception {
        String key = INODE_PREFIX + fileId;
        db.delete(key.getBytes(StandardCharsets.UTF_8));
    }

    @Override
    public void putDentry(String parentId, String name, String fileId) throws Exception {
        String key = DENTRY_PREFIX + parentId + ":" + name;
        db.put(key.getBytes(StandardCharsets.UTF_8), fileId.getBytes(StandardCharsets.UTF_8));
    }

    @Override
    public Optional<String> getDentry(String parentId, String name) throws Exception {
        String key = DENTRY_PREFIX + parentId + ":" + name;
        byte[] value = db.get(key.getBytes(StandardCharsets.UTF_8));
        if (value == null) {
            return Optional.empty();
        }
        return Optional.of(new String(value, StandardCharsets.UTF_8));
    }

    @Override
    public void deleteDentry(String parentId, String name) throws Exception {
        String key = DENTRY_PREFIX + parentId + ":" + name;
        db.delete(key.getBytes(StandardCharsets.UTF_8));
    }

    @Override
    public List<String> listDentries(String parentId) throws Exception {
        List<String> dentries = new ArrayList<>();
        String prefix = DENTRY_PREFIX + parentId + ":";
        byte[] prefixBytes = prefix.getBytes(StandardCharsets.UTF_8);
        try (RocksIterator iterator = db.newIterator()) {
            iterator.seek(prefixBytes);
            while (iterator.isValid()) {
                byte[] keyBytes = iterator.key();
                String key = new String(keyBytes, StandardCharsets.UTF_8);
                if (!key.startsWith(prefix)) {
                    break;
                }
                String name = key.substring(prefix.length());
                dentries.add(name);
                iterator.next();
            }
        }
        return dentries;
    }

    @Override
    public void putPathCache(String path, String fileId) throws Exception {
        String key = PATH_CACHE_PREFIX + path;
        db.put(key.getBytes(StandardCharsets.UTF_8), fileId.getBytes(StandardCharsets.UTF_8));
    }

    @Override
    public Optional<String> getPathCache(String path) throws Exception {
        String key = PATH_CACHE_PREFIX + path;
        byte[] value = db.get(key.getBytes(StandardCharsets.UTF_8));
        if (value == null) {
            return Optional.empty();
        }
        return Optional.of(new String(value, StandardCharsets.UTF_8));
    }

    @Override
    public void deletePathCache(String path) throws Exception {
        String key = PATH_CACHE_PREFIX + path;
        db.delete(key.getBytes(StandardCharsets.UTF_8));
    }

    @Override
    public void putBlock(String fileId, int blkIdx, BlockInfo blockInfo) throws Exception {
        String key = BLOCK_PREFIX + fileId + ":" + blkIdx;
        String value = objectMapper.writeValueAsString(blockInfo);
        db.put(key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8));
    }

    @Override
    public Optional<BlockInfo> getBlock(String fileId, int blkIdx) throws Exception {
        String key = BLOCK_PREFIX + fileId + ":" + blkIdx;
        byte[] value = db.get(key.getBytes(StandardCharsets.UTF_8));
        if (value == null) {
            return Optional.empty();
        }
        BlockInfo blockInfo = objectMapper.readValue(new String(value, StandardCharsets.UTF_8), BlockInfo.class);
        return Optional.of(blockInfo);
    }

    @Override
    public void deleteBlock(String fileId, int blkIdx) throws Exception {
        String key = BLOCK_PREFIX + fileId + ":" + blkIdx;
        db.delete(key.getBytes(StandardCharsets.UTF_8));
    }

    @Override
    public List<BlockInfo> getFileBlocks(String fileId) throws Exception {
        List<BlockInfo> allBlocks = new ArrayList<>();
        String prefix = BLOCK_PREFIX + fileId + ":";
        byte[] prefixBytes = prefix.getBytes(StandardCharsets.UTF_8);
        
        // 先收集所有块，按原始索引排序
        try (RocksIterator iterator = db.newIterator()) {
            iterator.seek(prefixBytes);
            while (iterator.isValid()) {
                byte[] keyBytes = iterator.key();
                String key = new String(keyBytes, StandardCharsets.UTF_8);
                if (!key.startsWith(prefix)) {
                    break;
                }
                byte[] valueBytes = iterator.value();
                String value = new String(valueBytes, StandardCharsets.UTF_8);
                BlockInfo blockInfo = objectMapper.readValue(value, BlockInfo.class);
                allBlocks.add(blockInfo);
                iterator.next();
            }
        }
        
        // 按原始块索引排序
        allBlocks.sort((a, b) -> Integer.compare(a.getBlkIdx(), b.getBlkIdx()));
        
        // 只返回有有效副本的块，保持原始索引
        List<BlockInfo> validBlocks = new ArrayList<>();
        for (BlockInfo blockInfo : allBlocks) {
            if (blockInfo.getReplicas() != null && !blockInfo.getReplicas().isEmpty()) {
                validBlocks.add(blockInfo);
                log.debug("Added valid block - index: {}, size: {}, replicas: {}", 
                         blockInfo.getBlkIdx(), blockInfo.getSize(), blockInfo.getReplicas().size());
            } else {
                log.debug("Skipped invalid block - index: {}, no replicas", blockInfo.getBlkIdx());
            }
        }
        
        log.info("Retrieved {} valid blocks out of {} total blocks for file {}", 
                validBlocks.size(), allBlocks.size(), fileId);
        return validBlocks;
    }

    @Override
    public List<BlockInfo> getAllBlocks() throws Exception {
        if (db == null) {
            log.warn("RocksDB not initialized, returning empty block list");
            return new ArrayList<>();
        }
        List<BlockInfo> blocks = new ArrayList<>();
        byte[] prefixBytes = BLOCK_PREFIX.getBytes(StandardCharsets.UTF_8);
        try (RocksIterator iterator = db.newIterator()) {
            iterator.seek(prefixBytes);
            while (iterator.isValid()) {
                byte[] keyBytes = iterator.key();
                String key = new String(keyBytes, StandardCharsets.UTF_8);
                if (!key.startsWith(BLOCK_PREFIX)) {
                    break;
                }
                byte[] valueBytes = iterator.value();
                String value = new String(valueBytes, StandardCharsets.UTF_8);
                BlockInfo blockInfo = objectMapper.readValue(value, BlockInfo.class);
                blocks.add(blockInfo);
                iterator.next();
            }
        }
        return blocks;
    }

    @Override
    public void putDataServer(DataServerInfo dsInfo) throws Exception {
        String key = DS_PREFIX + dsInfo.getDsId();
        String value = objectMapper.writeValueAsString(dsInfo);
        db.put(key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8));
    }

    @Override
    public Optional<DataServerInfo> getDataServer(String dsId) throws Exception {
        String key = DS_PREFIX + dsId;
        byte[] value = db.get(key.getBytes(StandardCharsets.UTF_8));
        if (value == null) {
            return Optional.empty();
        }
        DataServerInfo dsInfo = objectMapper.readValue(new String(value, StandardCharsets.UTF_8), DataServerInfo.class);
        return Optional.of(dsInfo);
    }

    @Override
    public void deleteDataServer(String dsId) throws Exception {
        String key = DS_PREFIX + dsId;
        db.delete(key.getBytes(StandardCharsets.UTF_8));
    }

    @Override
    public List<DataServerInfo> getAllDataServers() throws Exception {
        if (db == null) {
            log.warn("RocksDB not initialized, returning empty data server list");
            return new ArrayList<>();
        }
        List<DataServerInfo> dataServers = new ArrayList<>();
        byte[] prefixBytes = DS_PREFIX.getBytes(StandardCharsets.UTF_8);
        try (RocksIterator iterator = db.newIterator()) {
            iterator.seek(prefixBytes);
            while (iterator.isValid()) {
                byte[] keyBytes = iterator.key();
                String key = new String(keyBytes, StandardCharsets.UTF_8);
                if (!key.startsWith(DS_PREFIX)) {
                    break;
                }
                byte[] valueBytes = iterator.value();
                String value = new String(valueBytes, StandardCharsets.UTF_8);
                DataServerInfo dsInfo = objectMapper.readValue(value, DataServerInfo.class);
                dataServers.add(dsInfo);
                iterator.next();
            }
        }
        return dataServers;
    }

    @Override
    public void putClusterConfig(ClusterConfig config) throws Exception {
        String key = CONF_PREFIX + CLUSTER_KEY;
        String value = objectMapper.writeValueAsString(config);
        db.put(key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8));
    }

    @Override
    public Optional<ClusterConfig> getClusterConfig() throws Exception {
        String key = CONF_PREFIX + CLUSTER_KEY;
        byte[] value = db.get(key.getBytes(StandardCharsets.UTF_8));
        if (value == null) {
            return Optional.empty();
        }
        ClusterConfig config = objectMapper.readValue(new String(value, StandardCharsets.UTF_8), ClusterConfig.class);
        return Optional.of(config);
    }

    @Override
    public long getTotalInodeCount() throws Exception {
        checkDbInitialized();
        long count = 0;
        byte[] prefixBytes = INODE_PREFIX.getBytes(StandardCharsets.UTF_8);
        try (RocksIterator iterator = db.newIterator()) {
            iterator.seek(prefixBytes);
            while (iterator.isValid()) {
                byte[] keyBytes = iterator.key();
                String key = new String(keyBytes, StandardCharsets.UTF_8);
                if (!key.startsWith(INODE_PREFIX)) {
                    break;
                }
                count++;
                iterator.next();
            }
        }
        return count;
    }

    @Override
    public long getTotalBlockCount() throws Exception {
        checkDbInitialized();
        long count = 0;
        byte[] prefixBytes = BLOCK_PREFIX.getBytes(StandardCharsets.UTF_8);
        try (RocksIterator iterator = db.newIterator()) {
            iterator.seek(prefixBytes);
            while (iterator.isValid()) {
                byte[] keyBytes = iterator.key();
                String key = new String(keyBytes, StandardCharsets.UTF_8);
                if (!key.startsWith(BLOCK_PREFIX)) {
                    break;
                }
                count++;
                iterator.next();
            }
        }
        return count;
    }

    @Override
    public List<Inode> getAllInodes() throws Exception {
        checkDbInitialized();
        List<Inode> inodes = new ArrayList<>();
        byte[] prefixBytes = INODE_PREFIX.getBytes(StandardCharsets.UTF_8);
        try (RocksIterator iterator = db.newIterator()) {
            iterator.seek(prefixBytes);
            while (iterator.isValid()) {
                byte[] keyBytes = iterator.key();
                String key = new String(keyBytes, StandardCharsets.UTF_8);
                if (!key.startsWith(INODE_PREFIX)) {
                    break;
                }
                byte[] valueBytes = iterator.value();
                String value = new String(valueBytes, StandardCharsets.UTF_8);
                Inode inode = objectMapper.readValue(value, Inode.class);
                inodes.add(inode);
                iterator.next();
            }
        }
        return inodes;
    }

    @Override
    public String getInodePath(String fileId) throws Exception {
        checkDbInitialized();
        // 通过遍历路径缓存来查找对应的路径
        byte[] prefixBytes = PATH_CACHE_PREFIX.getBytes(StandardCharsets.UTF_8);
        try (RocksIterator iterator = db.newIterator()) {
            iterator.seek(prefixBytes);
            while (iterator.isValid()) {
                byte[] keyBytes = iterator.key();
                String key = new String(keyBytes, StandardCharsets.UTF_8);
                if (!key.startsWith(PATH_CACHE_PREFIX)) {
                    break;
                }
                byte[] valueBytes = iterator.value();
                String cachedFileId = new String(valueBytes, StandardCharsets.UTF_8);
                if (fileId.equals(cachedFileId)) {
                    // 找到匹配的路径
                    return key.substring(PATH_CACHE_PREFIX.length());
                }
                iterator.next();
            }
        }
        return null;
    }

    @Override
    public void clearAllData() throws Exception {
        checkDbInitialized();
        log.warn("清空所有元数据 - 这是一个危险操作！");
        
        // 删除所有数据
        String[] prefixes = {INODE_PREFIX, DENTRY_PREFIX, PATH_CACHE_PREFIX, BLOCK_PREFIX, DS_PREFIX, CONF_PREFIX};
        
        for (String prefix : prefixes) {
            byte[] prefixBytes = prefix.getBytes(StandardCharsets.UTF_8);
            try (RocksIterator iterator = db.newIterator()) {
                iterator.seek(prefixBytes);
                List<byte[]> keysToDelete = new ArrayList<>();
                
                while (iterator.isValid()) {
                    byte[] keyBytes = iterator.key();
                    String key = new String(keyBytes, StandardCharsets.UTF_8);
                    if (!key.startsWith(prefix)) {
                        break;
                    }
                    keysToDelete.add(keyBytes);
                    iterator.next();
                }
                
                // 批量删除
                for (byte[] key : keysToDelete) {
                    db.delete(key);
                }
                
                log.info("已清空前缀为 {} 的数据，共删除 {} 条记录", prefix, keysToDelete.size());
            }
        }
        
        log.warn("所有元数据已清空");
    }

//    @Override
//    public void backup(String backupPath) throws Exception {
//        // TODO: Implement backup using RocksDB checkpoint
//        throw new UnsupportedOperationException("Backup not implemented yet");
//    }
//
//    @Override
//    public void restore(String backupPath) throws Exception {
//        // TODO: Implement restore from backup
//        throw new UnsupportedOperationException("Restore not implemented yet");
//    }
}
